package com.tongcao.cn.thread;

import com.google.common.util.concurrent.ListenableFuture;
import com.google.common.util.concurrent.ListeningExecutorService;
import com.google.common.util.concurrent.MoreExecutors;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.*;

/**
 * @Author:tongcao
 * @Date:2018/8/23 17:52
 * Description:
 */
public enum ThreadPoolManager {

    /**
     * 实例
     */
    INSTANCE;

    /**
     * 线程池维护线程的最少数量
     */
    private int corePoolSize = 60;

    /**
     * 线程池维护线程的最大数量
     */
    private int maxPoolSize = 100;

    /**
     * 线程池维护线程所允许的空闲时间
     */
    private int keepAliveTime = 90;

    /**
     * 线程池所使用的缓冲队列大小
     */
    private int workQueueSize = 30;

    /**
     * 业务线程池
     */
    ListeningExecutorService bizPoolService;

    /**
     * 管道线程池
     */
    ListeningExecutorService channelPoolService;

    /**
     * 核心线程池
     */
    ListeningExecutorService corePoolService;

    private static final Logger logger = LoggerFactory.getLogger(ThreadPoolManager.class);

    /**
     * constructor
     */
    ThreadPoolManager() {

        /**
         * 业务线程池
         */
        ThreadPoolExecutor bizThreadPool = new ThreadPoolExecutor(corePoolSize, maxPoolSize, keepAliveTime, TimeUnit.SECONDS, new ArrayBlockingQueue<>(workQueueSize),
                new BookerThreadFactory("bizThreadPool"), new DiscardPolicy());
        bizPoolService = MoreExecutors.listeningDecorator(bizThreadPool);

        /**
         * 管道线程池
         */
        ThreadPoolExecutor channelThreadPool = new ThreadPoolExecutor(corePoolSize / 2, maxPoolSize / 2, keepAliveTime, TimeUnit.SECONDS, new ArrayBlockingQueue<>(workQueueSize),
                new BookerThreadFactory("channelPoolService"), new DiscardPolicy());
        channelPoolService = MoreExecutors.listeningDecorator(channelThreadPool);

        /**
         * 核心线程池
         */
        ThreadPoolExecutor coreThreadPool = new ThreadPoolExecutor(corePoolSize, maxPoolSize, keepAliveTime, TimeUnit.SECONDS, new ArrayBlockingQueue<>(workQueueSize),
                new BookerThreadFactory("coreThreadPool"), new DiscardPolicy());
        corePoolService = MoreExecutors.listeningDecorator(coreThreadPool);

    }


    /**
     * 拒绝策略添加日志
     */
    public class DiscardPolicy implements RejectedExecutionHandler {

        public DiscardPolicy() {
        }

        @Override
        public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
            logger.info("DiscardPolicy {} {}", r.toString(), e.getActiveCount());
        }
    }

    /**
     * 向线程池中添加任务方法(业务)
     */
    public <T> ListenableFuture<T> addBizExecuteTask(Callable<T> task) {
        return bizPoolService.submit(task);
    }

    /**
     * 向线程池中添加任务方法(核心)
     */
    public <T> ListenableFuture<T> addCoreExecuteTask(Callable<T> task) {
        return corePoolService.submit(task);
    }

    /**
     * 向线程池中添加任务方法(管道)
     */
    public <T> ListenableFuture<T> addChannelExecuteTask(Callable<T> task) {
        return channelPoolService.submit(task);
    }

    /**
     * @return
     */
    public ListeningExecutorService getBizPoolService() {
        return bizPoolService;
    }

    /**
     * @return
     */
    public ListeningExecutorService getCorePoolService() {
        return corePoolService;
    }

    /**
     * @return
     */
    public ListeningExecutorService getChannelPoolService() {
        return channelPoolService;
    }
}
